import org.apache.storm.spout.ISpout;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.io.*;
import java.util.Map;

/**
 * Created by shuaidong on 2017/7/23.
 */
public class MySpout implements IRichSpout{

    private FileInputStream fileInputStream;

    private BufferedReader bufferedReader;
    private  SpoutOutputCollector spoutOutputCollector;
    //初始化方式 map为配置文件
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {

        try {

            fileInputStream  = new FileInputStream("/root/track.log");
            bufferedReader = new BufferedReader(new InputStreamReader(fileInputStream));
            this.spoutOutputCollector = spoutOutputCollector;

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
    }

    public void close() {
        //进行资源释放
        try {
            bufferedReader.close();
            fileInputStream.close();

        } catch (IOException e) {
            e.printStackTrace();

        }

    }

    public void activate() {

    }

    public void deactivate() {

    }

    String string = null;
    public void nextTuple() {
        try {
            while ((string = bufferedReader.readLine()) !=null){
                //过滤动作
                this.spoutOutputCollector.emit(new Values(string));
            }

        } catch (Exception e) {
            e.printStackTrace();

        }


    }

    public void ack(Object o) {

        System.out.println("spout ack:"+o.toString());

    }

    public void fail(Object o) {
        System.out.println("spout fail:"+o.toString());

    }


    //定义输出字段
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("log"));

    }

    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

}
